package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo19Broadcast {
  def main(args: Array[String]): Unit = {
    /**
     * 广播变量
     */
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo18ACC")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    // 1、读取students、scores数据
    val stuRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")

    // 定义一个List
    val idsList: List[Int] = List(1500100001, 1500100011, 1500100021, 1500100031)

    // 如果需要在算子内部使用算子外部的变量，可以将变量进行广播
    // 将变量广播到每个Executor中 提高性能
    val idsListBro: Broadcast[List[Int]] = sc.broadcast(idsList)
    // 从stuRDD中过滤出 idsList中保存的学生
    // 结果没问题 但这里会出现性能问题
    // 算子内部的代码最终会被封装成Task发送到Executor中执行
    // 每个Task中会封装idsList的一个副本
    // Task最终都是在Executor中执行=>>没有必要每个Task放一个副本，可以在每个Executor上放一个副本
    // 减少分发Task时网络的开销，提高效率
    stuRDD.filter(line => {
      // 提取学生id
      val id: Int = line.split(",")(0).toInt
      // 通过id进行过滤
      // 也使用了外部变量 但没有进行修改(不会生效)
      //      idsList.contains(id)
      // 使用广播变量获取算子外部定义的变量
      idsListBro.value.contains(id)
    }).foreach(println)


  }

}
